use crate::config::data_source_config::{DataSourceConfig, DataSourceManager};
use arrow::array::{ArrayRef, Float64Array, TimestampNanosecondArray};
use arrow::datatypes::{DataType, Field, Schema};
use arrow::record_batch::RecordBatch;
use chrono::{DateTime, Utc};
use futures::stream::StreamExt;
use serde::{Deserialize, Serialize};
use std::sync::Arc;
use tokio::sync::{broadcast, RwLock};
use tokio::time::{Duration, Instant};

#[repr(C)]
struct QuoteData {
    price: f64,
    volume: f64,
    amount: f64,
    bid_price: f64,
    ask_price: f64,
    bid_volume: f64,
    ask_volume: f64,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MarketData {
    pub symbol: String,
    pub timestamp: DateTime<Utc>,
    pub price: f64,
    pub volume: f64,
    pub amount: f64,
    pub bid_price: f64,
    pub ask_price: f64,
    pub bid_volume: f64,
    pub ask_volume: f64,
}

use std::collections::HashMap;
use std::time::SystemTime;

// 缓存项结构
struct CacheItem<T> {
    data: T,
    expires_at: SystemTime,
}

impl<T> CacheItem<T> {
    fn new(data: T, ttl_secs: u64) -> Self {
        Self {
            data,
            expires_at: SystemTime::now() + std::time::Duration::from_secs(ttl_secs),
        }
    }
    
    fn is_expired(&self) -> bool {
        SystemTime::now() > self.expires_at
    }
}

#[derive(Clone)]
pub struct MarketDataProcessor {
    data_source_manager: Arc<DataSourceManager>,
    data_tx: broadcast::Sender<MarketData>,
    // 添加数据缓存
    data_cache: Arc<RwLock<HashMap<String, CacheItem<MarketData>>>>,
    cache_ttl: u64, // 缓存有效期（秒）
}

impl MarketDataProcessor {
    pub fn new(data_source_manager: Arc<DataSourceManager>) -> Self {
        let (data_tx, _) = broadcast::channel(1000); // 缓冲1000条数据
        let processor = Self {
            data_source_manager,
            data_tx,
            data_cache: Arc::new(RwLock::new(HashMap::new())),
            cache_ttl: 5, // 默认5秒缓存
        };
        
        // 启动缓存清理任务
        let processor_clone = processor.clone();
        tokio::spawn(async move {
            let mut interval = tokio::time::interval(Duration::from_secs(60)); // 每分钟清理一次
            loop {
                interval.tick().await;
                processor_clone.clear_expired_cache().await;
            }
        });
        
        processor
    }

    pub async fn start_collection(&self, symbols: Vec<String>, interval_ms: u64) {
        let mut interval = tokio::time::interval(Duration::from_millis(interval_ms));
        let mut last_switch = Instant::now();
        let mut using_primary = true;

        loop {
            interval.tick().await;

            // 获取数据源配置
            let source_config = if using_primary {
                self.data_source_manager.get_primary_source().await
            } else {
                self.data_source_manager.get_backup_source().await
            };

            let Some(config) = source_config else {
                log::error!("No available data source");
                continue;
            };

            // 批量获取数据
            let result = self.fetch_batch_data(&config, &symbols).await;
            
            match result {
                Ok(data_batch) => {
                    // 发送数据到通道
                    for data in data_batch {
                        if let Err(e) = self.data_tx.send(data) {
                            log::error!("Failed to send market data: {}", e);
                        }
                    }
                    // 如果是使用备用源，每隔一段时间尝试切回主源
                    if !using_primary && last_switch.elapsed() > Duration::from_secs(300) {
                        using_primary = true;
                        last_switch = Instant::now();
                    }
                }
                Err(e) => {
                    log::error!("Failed to fetch market data: {}", e);
                    if using_primary {
                        // 切换到备用源
                        using_primary = false;
                        last_switch = Instant::now();
                    }
                }
            }
        }
    }

    async fn fetch_batch_data(
        &self,
        config: &DataSourceConfig,
        symbols: &[String],
    ) -> Result<Vec<MarketData>, Box<dyn std::error::Error>> {
        // 将symbols分为两组：缓存中有效的和需要重新获取的
        let mut result_data = Vec::new();
        let mut symbols_to_fetch = Vec::new();
        
        // 检查缓存
        {
            let cache = self.data_cache.read().await;
            for symbol in symbols {
                if let Some(cache_item) = cache.get(symbol) {
                    if !cache_item.is_expired() {
                        result_data.push(cache_item.data.clone());
                        continue;
                    }
                }
                symbols_to_fetch.push(symbol.clone());
            }
        }
        
        // 如果所有数据都在缓存中，则直接返回
        if symbols_to_fetch.is_empty() {
            return Ok(result_data);
        }
        
        // 获取未缓存的数据
        let fresh_data = match config.source_type.as_str() {
            "tdx" => self.fetch_from_tdx(config, &symbols_to_fetch).await?,
            "akshare" => self.fetch_from_akshare(config, &symbols_to_fetch).await?,
            _ => return Err("Unsupported data source type".into()),
        };
        
        // 更新缓存
        {
            let mut cache = self.data_cache.write().await;
            for data in &fresh_data {
                cache.insert(
                    data.symbol.clone(),
                    CacheItem::new(data.clone(), self.cache_ttl),
                );
            }
        }
        
        // 合并结果
        result_data.extend(fresh_data);
        Ok(result_data)
    }

    async fn fetch_from_tdx(
        &self,
        config: &DataSourceConfig,
        symbols: &[String],
    ) -> Result<Vec<MarketData>, Box<dyn std::error::Error>> {
        use std::path::Path;
        use chrono::Local;
        use libloading::{Library, Symbol};

        // 获取通达信DLL路径
        let tdx_path = match config.connection_params.get("path") {
            Some(path) => path.as_str().ok_or("Invalid TDX path")?,
            None => return Err("TDX path not configured".into()),
        };
        let library_name = match config.connection_params.get("library") {
            Some(lib) => lib.as_str().ok_or("Invalid library name")?,
            None => return Err("TDX library not configured".into()),
        };

        let library_path = Path::new(tdx_path).join(library_name);
        
        // 加载通达信DLL
        let lib = unsafe { Library::new(library_path)? };

        // 定义市场数据缓存
        let mut market_data = Vec::with_capacity(symbols.len());
        
        for symbol in symbols {
            // 解析股票代码，获取市场类型
            let market_type = if symbol.starts_with('6') { 1 } else { 0 }; // 1为上海，0为深圳
            
            // 获取实时行情
            type GetQuoteFn = unsafe extern "C" fn(market: i32, code: *const i8) -> *const QuoteData;
            let get_quote: Symbol<GetQuoteFn> = unsafe { lib.get(b"TdxGetQuote\0")? };
            
            // 将股票代码转换为C字符串
            let code = std::ffi::CString::new(symbol.as_bytes())?;
            
            // 调用DLL获取行情数据
            let quote_data = unsafe { get_quote(market_type, code.as_ptr()) };
            if quote_data.is_null() {
                log::warn!("Failed to get quote data for symbol: {}", symbol);
                continue;
            }
            
            // 将C结构体转换为Rust结构体
            let quote = unsafe { &*quote_data };
            
            // 构造MarketData
            market_data.push(MarketData {
                symbol: symbol.clone(),
                timestamp: Local::now().into(),
                price: quote.price,
                volume: quote.volume,
                amount: quote.amount,
                bid_price: quote.bid_price,
                ask_price: quote.ask_price,
                bid_volume: quote.bid_volume,
                ask_volume: quote.ask_volume,
            });
        }

        Ok(market_data)
    }
    }

    async fn fetch_from_akshare(
        &self,
        config: &DataSourceConfig,
        symbols: &[String],
    ) -> Result<Vec<MarketData>, Box<dyn std::error::Error>> {
        use pyo3::{PyResult, Python, types::PyModule};
        use chrono::Local;

        // 获取Python GIL
        let gil = Python::acquire_gil();
        let py = gil.python();
        
        // 导入akshare模块
        let akshare = PyModule::import(py, "akshare")?;
        
        // 定义市场数据缓存
        let mut market_data = Vec::with_capacity(symbols.len());
        
        for symbol in symbols {
            // 根据股票代码确定市场类型
            let market = if symbol.starts_with('6') { "sh" } else { "sz" };
            let full_symbol = format!("{}{}", market, symbol);
            
            // 调用akshare的股票实时行情接口
            let kwargs = pyo3::types::PyDict::new(py);
            kwargs.set_item("symbol", &full_symbol)?;
            
            let quote_df = akshare.getattr("stock_zh_a_spot_em")?.call((), Some(kwargs))?;
            
            // 如果返回的DataFrame为空，则跳过
            if quote_df.getattr("empty")?.extract::<bool>()? {
                log::warn!("Empty quote data for symbol: {}", symbol);
                continue;
            }
            
            // 提取所需数据
            let price = quote_df.getattr("iloc")?.call1((0,))?.getattr("getitem")?.call1(("最新价",))?.extract::<f64>()?;
            let volume = quote_df.getattr("iloc")?.call1((0,))?.getattr("getitem")?.call1(("成交量",))?.extract::<f64>()?;
            let amount = quote_df.getattr("iloc")?.call1((0,))?.getattr("getitem")?.call1(("成交额",))?.extract::<f64>()?;
            let bid_price = quote_df.getattr("iloc")?.call1((0,))?.getattr("getitem")?.call1(("买一价",))?.extract::<f64>()?;
            let ask_price = quote_df.getattr("iloc")?.call1((0,))?.getattr("getitem")?.call1(("卖一价",))?.extract::<f64>()?;
            let bid_volume = quote_df.getattr("iloc")?.call1((0,))?.getattr("getitem")?.call1(("买一量",))?.extract::<f64>()?;
            let ask_volume = quote_df.getattr("iloc")?.call1((0,))?.getattr("getitem")?.call1(("卖一量",))?.extract::<f64>()?;
            
            // 构造MarketData
            market_data.push(MarketData {
                symbol: symbol.clone(),
                timestamp: Local::now().into(),
                price,
                volume,
                amount,
                bid_price,
                ask_price,
                bid_volume,
                ask_volume,
            });
        }

        Ok(market_data)
    }
    }

    pub fn subscribe(&self) -> broadcast::Receiver<MarketData> {
        self.data_tx.subscribe()
    }
    
    // 设置缓存TTL
    pub fn set_cache_ttl(&mut self, ttl_secs: u64) {
        self.cache_ttl = ttl_secs;
    }
    
    // 清理过期缓存
    pub async fn clear_expired_cache(&self) {
        let mut cache = self.data_cache.write().await;
        cache.retain(|_, item| !item.is_expired());
    }
    
    // 完全清空缓存
    pub async fn clear_all_cache(&self) {
        let mut cache = self.data_cache.write().await;
        cache.clear();
    }

    // 将市场数据转换为Arrow格式
    pub fn to_arrow_batch(data: &[MarketData]) -> Result<RecordBatch, Box<dyn std::error::Error>> {
        let schema = Schema::new(vec![
            Field::new("timestamp", DataType::Timestamp(arrow::datatypes::TimeUnit::Nanosecond, None), false),
            Field::new("price", DataType::Float64, false),
            Field::new("volume", DataType::Float64, false),
            Field::new("amount", DataType::Float64, false),
            Field::new("bid_price", DataType::Float64, false),
            Field::new("ask_price", DataType::Float64, false),
            Field::new("bid_volume", DataType::Float64, false),
            Field::new("ask_volume", DataType::Float64, false),
        ]);

        let timestamps: Vec<_> = data.iter().map(|d| d.timestamp.timestamp_nanos()).collect();
        let prices: Vec<_> = data.iter().map(|d| d.price).collect();
        let volumes: Vec<_> = data.iter().map(|d| d.volume).collect();
        let amounts: Vec<_> = data.iter().map(|d| d.amount).collect();
        let bid_prices: Vec<_> = data.iter().map(|d| d.bid_price).collect();
        let ask_prices: Vec<_> = data.iter().map(|d| d.ask_price).collect();
        let bid_volumes: Vec<_> = data.iter().map(|d| d.bid_volume).collect();
        let ask_volumes: Vec<_> = data.iter().map(|d| d.ask_volume).collect();

        let arrays: Vec<ArrayRef> = vec![
            Arc::new(TimestampNanosecondArray::from(timestamps)),
            Arc::new(Float64Array::from(prices)),
            Arc::new(Float64Array::from(volumes)),
            Arc::new(Float64Array::from(amounts)),
            Arc::new(Float64Array::from(bid_prices)),
            Arc::new(Float64Array::from(ask_prices)),
            Arc::new(Float64Array::from(bid_volumes)),
            Arc::new(Float64Array::from(ask_volumes)),
        ];

        Ok(RecordBatch::try_new(Arc::new(schema), arrays)?)
    }
}